package com.atguigu.tingshu.common.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Author:huzhongkui
 * Date: 2024-03-28  23:06
 */

@Component
public class KafkaService {



    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

//    /**
//     * 发送消息
//     * @param topic
//     * @param value
//     * @return
//     */
//    public boolean sendMessage(String topic, Object value) {
//        return this.sendMessage(topic, null, value);
//    }
//    /**
//     * 发送消息
//     * @param topic
//     * @param key
//     * @param value
//     * @return
//     */
//    public boolean sendMessage(String topic, String key, Object value) {
//        CompletableFuture<SendResult<String, Object>> completableFuture = kafkaTemplate.send(topic, key, value);
//        //执行成功回调
//        completableFuture.thenAccept(result -> {
//            //System.out.println("发送成功:{}" + JSON.toJSONString(result));
//            logger.debug("kafka发送消息成功: topic={}, key={}, value={}", topic, key, JSON.toJSONString(value));
//        });
//        //执行失败回调
//        completableFuture.exceptionally(e -> {
//            //发送失败，记录日志或者采取重发策略
//            logger.error("kafka发送消息失败: topic={}, key={}, value={}", topic, key, JSON.toJSONString(value));
//            e.printStackTrace();
//            return null;
//        });
//        return true;
//    }
//
//
//    /**
//     * 延迟消息
//     * @param topic
//     * @param value
//     * @param delayTime 延迟的时间，单位：秒
//     * @return
//     */
//    public boolean sendDelayMessage(String topic, Object value, int delayTime) {
//        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
//        executor.schedule(() -> kafkaTemplate.send(topic, value), delayTime, TimeUnit.SECONDS);
//        executor.shutdown();
//        return false;
//    }


    /**
     * 发送消息(不带路由键)
     */
    public void sendMsg(String topic, Object value) {
        CompletableFuture send = kafkaTemplate.send(topic, value);
        send.whenComplete((o, e) -> {
            if (e == null) {
                logger.info("发送消息:{}成功", value);
            } else {
                logger.info("发送消息:{}失败", JSONObject.toJSONString(value));
            }
        });
    }

    /**
     * 发送消息（带路由键）
     */

    public void sendMsg(String topic, String key, Object value) {
        CompletableFuture send = kafkaTemplate.send(topic, key, value);
        send.whenComplete((o, e) -> {
            if (e == null) {
                logger.info("发送消息:{},失败，使用的路由键:{}", JSONObject.toJSONString(value), key);
            } else {
                logger.info("发送消息:{}成功,使用的路由键:{}", JSONObject.toJSONString(value), key);
            }
        });
    }

    /**
     * 发送消息带分区+带路由键
     */

    public void sendMsg(String topic, Integer partition, String key, Object value) {
        CompletableFuture send = kafkaTemplate.send(topic, partition, key, value);
        send.whenComplete((o, throwable) -> {
            if (throwable != null) {
                logger.info("发送消息:{},进入到分区:{}失败,使用的路由键:{}", JSONObject.toJSONString(value), partition, key);
            } else {
                logger.info("发送消息:{},进入到分区:{}成功", JSONObject.toJSONString(value), partition, value);
            }
        });
    }

}
